package com.gitee.cui.concurrency.designmodel.masterworker;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * Master-Worker模式中的Worker角色，负责具体的业务处理
 * @author cuiqingqiang
 * @version 1.0
 * @date 2021/12/8 16:10
 **/
public class Worker<T extends Task, R> {

    /**
     * 接收任务的阻塞队列
     */
    private LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();

    /**
     * 工作线程id
     */
    private static final AtomicInteger index = new AtomicInteger(1);

    /**
     * 工作者id
     */
    private int workerId;

    /**
     * 工作线程
     */
    private Thread thread;

    public Worker() {
        this.workerId = index.getAndIncrement();
        this.thread = new Thread(this::run);
        thread.start();
    }

    /**
     * 执行具体任务的逻辑
     */
    public void run() {
        // 循环执行任务
        while (true) {
            // 从任务队列中获取一个任务并执行
            try {
                T task = this.taskQueue.take();
                task.setWorkerId(workerId);
                task.execute();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 向任务队列中提交任务和回调函数
     * @param task 任务
     * @param action 回调函数
     */
    public void submit(T task, Consumer<R> action) {
        task.setResultAction(action);
        try {
            this.taskQueue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
